package org.wyz.mapreduce.pagerank;

import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;
import java.io.IOException;

public class PageRankReducer extends Reducer<Text, PageWritable, Text, NullWritable> {

    // 网页数量
    private static final int N_PAGE = 4;

    // 超参数ratio
    private static final double RATIO = 0.85;

    @Override
    protected void reduce(Text key, Iterable<PageWritable> values, Context context) throws IOException, InterruptedException {

        double newRank = 0;

        String structure = "";
        for (PageWritable v : values) {
            if (v.tag == 0) {
                double inRank = Double.parseDouble(v.content);
                newRank += inRank;
            } else if (v.tag == 1) {
                structure = v.content;
            }

        }
        newRank = newRank * RATIO + (1 - RATIO) / N_PAGE;
        String[] ss = structure.split("\\s");
        ss[0] = String.valueOf(newRank);
        String wholeKey = key.toString() + " " + String.join(" ", ss);

        context.write(new Text(wholeKey), NullWritable.get());
    }
}
